package cn.iotab.iotkit.transport.mqtt.handler;

import cn.iotab.iotkit.transport.api.common.message.msg.CommonPayload;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionInfo;
import cn.iotab.iotkit.transport.api.common.queue.SendResultCallback;
import cn.iotab.iotkit.transport.mqtt.MqttTransportContext;
import cn.iotab.iotkit.transport.mqtt.adaptors.AdaptorException;
import cn.iotab.iotkit.transport.mqtt.adaptors.MqttTransportAdaptor;
import cn.iotab.iotkit.transport.mqtt.session.MqttSessionCtx;
import cn.iotab.iotkit.transport.mqtt.topic.MqttTopics;
import cn.iotab.iotkit.transport.mqtt.topic.TopicMeta;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

@Slf4j
public class PublishHandler extends AbstractMqttMessageHandler<MqttPublishMessage>{


    public PublishHandler(MqttTransportContext context, MqttSessionCtx deviceSessionCtx) {
        super(context, deviceSessionCtx);
    }

    @Override
    public void process(ChannelHandlerContext ctx, MqttPublishMessage mqttMsg) {
        if (!checkConnected(ctx, mqttMsg)) {
            return;
        }
        String topicName = mqttMsg.variableHeader().topicName();
        int msgId = mqttMsg.variableHeader().packetId();
        log.trace("[{}] Processing publish msg [{}][{}]!", sessionId, topicName, msgId);

        processDevicePublish(ctx, mqttMsg, topicName, msgId);

    }

    private void processDevicePublish(ChannelHandlerContext ctx, MqttPublishMessage mqttMsg, String topicName, int msgId) {
        TopicMeta topicMeta = MqttTopics.parseTopic(topicName);
        //校验productKey与deviceID合法性
        SessionInfo sessionInfo = deviceSessionCtx.getSessionInfo();
        if (topicMeta.isValid()) {
            String productKey = topicMeta.getProductKey();
            String deviceId = topicMeta.getDeviceId();
            if (!StringUtils.equals(productKey, sessionInfo.getProductKey())
                    || !StringUtils.equals(deviceId, sessionInfo.getDeviceId())) {
                log.error("[{}] Failed to publish message to [{}], please see correct productKey:{}, deviceId:{}",
                        sessionId, topicName, sessionInfo.getProductKey(), sessionInfo.getDeviceId());
                ctx.close();
                return;
            }
        }
        try {
            MqttTransportAdaptor payloadAdaptor = deviceSessionCtx.getPayloadAdaptor();
            switch (topicMeta.getTopicType()){
                case DEVICE_POST_PLATFORM:
                    CommonPayload rpcRequestMsg = payloadAdaptor.convertFromDeviceRequest(topicMeta.getProductKey(), topicMeta.getDeviceId(), mqttMsg);
                    transportDriverService.onDeviceRequestMessage(deviceSessionCtx.getSessionInfo(), rpcRequestMsg, getPubAckCallback(ctx, msgId, rpcRequestMsg));
                    break;
                case DEVICE_REPLY_PLATFORM:
                    CommonPayload claimDeviceMsg = payloadAdaptor.convertFromDeviceResponse(topicMeta.getProductKey(), topicMeta.getDeviceId(), mqttMsg);
                    transportDriverService.onDeviceResponseMessage(deviceSessionCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(ctx, msgId, claimDeviceMsg));
                    break;
                default:
                    log.warn("invalid topic {}, close channel", topicName);
                    ctx.close();
            }
        } catch (AdaptorException e) {
            log.warn("[{}] Failed to process publish msg [{}][{}]", sessionId, topicName, msgId, e);
            log.info("[{}] Closing current session due to invalid publish msg [{}][{}]", sessionId, topicName, msgId);
            ctx.close();
        }
    }

    private void ack(ChannelHandlerContext ctx, int msgId) {
        if (msgId > 0) {
            ctx.writeAndFlush(createMqttPubAckMsg(msgId));
        }
    }

    public static MqttPubAckMessage createMqttPubAckMsg(int requestId) {
        MqttFixedHeader mqttFixedHeader =
                new MqttFixedHeader(MqttMessageType.PUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessageIdVariableHeader mqttMsgIdVariableHeader =
                MqttMessageIdVariableHeader.from(requestId);
        return new MqttPubAckMessage(mqttFixedHeader, mqttMsgIdVariableHeader);
    }

    private <T> SendResultCallback<Void> getPubAckCallback(final ChannelHandlerContext ctx, final int msgId, final T msg) {
        return new SendResultCallback<Void>() {
            @Override
            public void onSuccess(Void dummy) {
                log.trace("[{}] Published msg: {}", sessionId, msg);
                ack(ctx, msgId);
            }

            @Override
            public void onError(Throwable e) {
                log.trace("[{}] Failed to publish msg: {}", sessionId, msg, e);
                deviceSessionCtx.processDisconnect();
            }
        };
    }
}
